package com.hillky.core;

import com.hillky.NettyBootstrapInitializer;
import com.hillky.YrpcBootstrap;
import com.hillky.compress.CompressorFactory;
import com.hillky.discovery.Registry;
import com.hillky.enumeration.RequestType;
import com.hillky.serialize.SerializerFactory;
import com.hillky.transport.message.YrpcRequest;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFutureListener;
import lombok.extern.slf4j.Slf4j;

import java.net.InetSocketAddress;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * 我们的核心目的是：感知哪些服务器的连接状态是正常和不正常的
 * 心跳检测器
 */
@Slf4j
public class HeartbeatDetector {

    public static void detectHeartbeat(String serviceName){
        // 从注册中心拉取服务列表并建立连接
        Registry registry = YrpcBootstrap.getInstance().getConfiguration().getRegistryConfig().getRegistry();
        List<InetSocketAddress> addresses = registry.lookup(serviceName,YrpcBootstrap.getInstance().getConfiguration().getGroup());

        // 将连接进行缓存
        for (InetSocketAddress address : addresses) {
            try {
                if(!YrpcBootstrap.CHANNEL_CACHE.containsKey(address)){
                    Channel channel = NettyBootstrapInitializer.getBootstrap().connect(address).sync().channel();
                    YrpcBootstrap.CHANNEL_CACHE.put(address,channel);
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

        // 任务：定期发送消息
        // 使用线程池进行
        Thread thread = new Thread(() -> {
            Timer timer = new Timer();
            timer.scheduleAtFixedRate(new myTimerTask(), 0, 2000);
        },"yrpc-HeartbeatDetector");
        thread.setDaemon(true);
        thread.start();


    }

    private static class myTimerTask extends TimerTask{

        @Override
        public void run() {

            // 将响应市长的map清空
            YrpcBootstrap.ANSWER_TIME_CHANNEL_CACHE.clear();

            // 遍历所有channel
            Map<InetSocketAddress, Channel> cache = YrpcBootstrap.CHANNEL_CACHE;
            for (Map.Entry<InetSocketAddress, Channel> entry : cache.entrySet()) {

                int tryTimes=3;

                while (tryTimes>0){
                    Channel channel = entry.getValue();
                    long start=System.currentTimeMillis();

                    // 构建一个心跳请求
                    YrpcRequest yrpcRequest = YrpcRequest.builder()
                            .requestId(YrpcBootstrap.getInstance().getConfiguration().getIdGenerator().getId())
                            .compressType(CompressorFactory.getCompressor(YrpcBootstrap.getInstance().getConfiguration().getCompressType()).getCode())
                            .requestType(RequestType.HEART_BEAT.getId())
                            .serializeType(SerializerFactory.getSerializer(YrpcBootstrap.getInstance().getConfiguration().getSerializeType()).getCode())
                            .timeStamp(start)
                            .build();

                    //4.要写出报文
                    CompletableFuture<Object> completableFuture=new CompletableFuture<>();

                    //暴将露出去
                    YrpcBootstrap.PENDING_REQUEST.put(yrpcRequest.getRequestId(),completableFuture);

                    channel.writeAndFlush(yrpcRequest).addListener((ChannelFutureListener) promise -> {
                        if(!promise.isSuccess()){
                            completableFuture.completeExceptionally(promise.cause());
                        }
                    });

                    Long endTime=null;
                    try {
                        completableFuture.get(1, TimeUnit.SECONDS);
                        endTime= System.currentTimeMillis();
                    } catch (InterruptedException | TimeoutException | ExecutionException e) {
                        // 一旦发生问题，需要优先重试
                        tryTimes--;

                        log.error("和地址为【{}】的主机连接发生异常,正在进行第【{}】次重试...",channel.remoteAddress(),3-tryTimes);

                        if(tryTimes==0){
                            // 将失效的地址移出服务列表
                            YrpcBootstrap.CHANNEL_CACHE.remove(entry.getKey());
                        }

                        try {
                            Thread.sleep( 10 *(new Random().nextInt(5)));
                        } catch (InterruptedException interruptedException) {
                            interruptedException.printStackTrace();
                        }
                        continue;
                    }

                    Long time=endTime-start;

                    // 使用TreeMap进行缓存
                    YrpcBootstrap.ANSWER_TIME_CHANNEL_CACHE.put(time,channel);

                    log.debug("和【{}】服务器的响应时间是【{}】",entry.getKey(),time);

                    break;
                }

            }


            log.info("-----------------------------响应时间的treemap-----------------------------");
            for (Map.Entry<Long, Channel> entry : YrpcBootstrap.ANSWER_TIME_CHANNEL_CACHE.entrySet()) {
                if(log.isDebugEnabled()){
                    log.debug("[{}]---->channelID:[{}]",entry.getKey(),entry.getValue().id());
                }
            }


        }
    }

}
